package com.ice.remoting;

import java.net.InetSocketAddress;
import java.util.concurrent.*;

import com.ice.framework.common.exception.BusinessException;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.commons.lang3.exception.ExceptionUtils;

import com.ice.framework.common.util.AssertException;
import com.ice.remoting.code.DecoderFactory;
import com.ice.remoting.code.EncodeFactory;
import com.ice.remoting.common.NetworkUtil;
import com.ice.remoting.common.RemotingHelper;
import com.ice.remoting.common.ThreadFactoryImpl;
import com.ice.remoting.common.ThreadUtils;
import com.ice.remoting.data.DeviceLiveInfo;
import com.ice.remoting.data.RemotingCommand;
import com.ice.remoting.exception.RemotingConnectException;
import com.ice.remoting.exception.RemotingSendRequestException;
import com.ice.remoting.exception.RemotingTimeoutException;
import com.ice.remoting.listener.ChannelEventListener;
import com.ice.remoting.listener.DefaultChannelEventListener;
import com.ice.remoting.netty.*;

import cn.hutool.core.lang.Pair;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.extern.slf4j.Slf4j;

/**
 * @author wangwei @Date 2024/1/3 10:38
 */
@Slf4j
public abstract class NettyRemotingServer extends NettyRemotingAbstract implements DecoderFactory, EncodeFactory {

    /**
     * 配置
     */
    protected NettyServerConfig nettyServerConfig;

    private ServerBootstrap serverBootstrap;
    private final EventLoopGroup eventLoopGroupSelector;
    private final EventLoopGroup eventLoopGroupBoss;
    /**
     * 事件监听器
     */
    private final ChannelEventListener channelEventListener;
    /**
     * 连接管理器
     */
    private NettyConnectManageHandler connectionManageHandler;
    /**
     * 任务处理器
     */
    private NettyServerHandler nettyServerHandler;
    /**
     * 设备管理器
     */
    private DeviceManager deviceManager;

    private final ExecutorService publicExecutor;

    private final HashedWheelTimer timer = new HashedWheelTimer(r -> new Thread(r, "ServerHouseKeepingService"));

    private final ScheduledExecutorService scanExecutorService = ThreadUtils.newScheduledThreadPool(1,
        new BasicThreadFactory.Builder().namingPattern("NSScanScheduledThread").daemon(true).build());

    private ExecutorService clientRequestExecutor;
    private BlockingQueue<Runnable> clientRequestThreadPoolQueue;

    public NettyRemotingServer(NettyServerConfig nettyServerConfig) {
        this(nettyServerConfig, null);
    }

    public NettyRemotingServer(NettyServerConfig nettyServerConfig, ChannelEventListener channelEventListener) {
        this.nettyServerConfig = nettyServerConfig;
        AssertException.isNotEmpty(this.nettyServerConfig.getListenPort(),
            "需要继承NettyServerConfig，并实现getListenPort()方法");
        this.serverBootstrap = new ServerBootstrap();
        this.eventLoopGroupBoss = buildBossEventLoopGroup();
        this.eventLoopGroupSelector = buildEventLoopGroupSelector();

        this.connectionManageHandler = new NettyConnectManageHandler();
        this.channelEventListener =
            channelEventListener != null ? channelEventListener : new DefaultChannelEventListener(nettyServerConfig);
        this.deviceManager = new DeviceManager(this.nettyServerConfig);
        this.deviceManager.initialize();
        this.channelEventListener.setManagerRouter(deviceManager);
        this.nettyServerHandler = new NettyServerHandler();
        this.publicExecutor = buildPublicExecutor(nettyServerConfig);

        this.clientRequestThreadPoolQueue =
            new LinkedBlockingQueue<>(this.nettyServerConfig.getClientRequestThreadPoolQueueCapacity());
        this.clientRequestExecutor =
            ThreadUtils.newThreadPoolExecutor(nettyServerConfig.getClientRequestThreadPoolNums(),
                this.nettyServerConfig.getClientRequestThreadPoolNums(), 1000 * 60, TimeUnit.MILLISECONDS,
                this.clientRequestThreadPoolQueue, new ThreadFactoryImpl("ClientRequestExecutorThread_"));
    }

    public void start() {
        this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
            .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 1024).option(ChannelOption.SO_REUSEADDR, true)
            .childOption(ChannelOption.SO_KEEPALIVE, false).childOption(ChannelOption.TCP_NODELAY, true)
            .localAddress(
                new InetSocketAddress(this.nettyServerConfig.getBindAddress(), this.nettyServerConfig.getListenPort()))
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel channel) throws Exception {
                    configChannel(channel);
                }
            });
        try {
            ChannelFuture sync = serverBootstrap.bind().sync();
            log.info("RemotingServer started, listening {}:{}", this.nettyServerConfig.getBindAddress(),
                this.nettyServerConfig.getListenPort());
        } catch (Exception e) {
            throw new IllegalStateException(String.format("Failed to bind to %s:%d", nettyServerConfig.getBindAddress(),
                nettyServerConfig.getListenPort()), e);
        }
        if (this.channelEventListener != null) {
            this.nettyEventExecutor.start();
        }

        // 扫描ResponseTable类,超时执行
        TimerTask timerScanResponseTable = new TimerTask() {
            @Override
            public void run(Timeout timeout) {
                try {
                    NettyRemotingServer.this.scanResponseTable();
                } catch (Throwable e) {
                    log.error("scanResponseTable exception", e);
                } finally {
                    timer.newTimeout(this, 1000, TimeUnit.MILLISECONDS);
                }
            }
        };
        this.timer.newTimeout(timerScanResponseTable, 1000 * 3, TimeUnit.MILLISECONDS);
        if (nettyServerConfig.isScanBrokerFlag()) {
            this.deviceManager.start();
        }
    }

    public void stop() {
        this.eventLoopGroupBoss.shutdownGracefully();
        this.eventLoopGroupSelector.shutdownGracefully();
        if (nettyServerConfig.isScanBrokerFlag()) {
            this.deviceManager.shutdown();
        }
        // 事件关闭
        this.nettyEventExecutor.shutdown();
        clientRequestExecutor.shutdown();
        clearRPCHook();
    }

    public void registerProcessor(int requestCode, NettyRequestProcessor processor, ExecutorService executor) {
        ExecutorService executorThis = executor;
        if (executor == null) {
            executorThis = publicExecutor;
        }
        Pair<NettyRequestProcessor, ExecutorService> pair = new Pair<>(processor, executorThis);
        this.processorTable.put(requestCode, pair);
    }

    public void registerDefaultProcessor(NettyRequestProcessor processor) {
        this.defaultRequestProcessorPair = new Pair<>(processor, this.clientRequestExecutor);
    }

    protected ChannelPipeline configChannel(SocketChannel ch) {
        return ch.pipeline().addLast(new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()),
            this.buildDecoder(), // 解码
            this.buildEncoder(), // 编码
            this.connectionManageHandler, //
            this.nettyServerHandler); //
    }

    private EventLoopGroup buildEventLoopGroupSelector() {
        if (useEpoll()) {
            return new EpollEventLoopGroup(nettyServerConfig.getServerSelectorThreads(),
                new ThreadFactoryImpl("NettyServerEPOLLSelector_"));
        } else {
            return new NioEventLoopGroup(nettyServerConfig.getServerSelectorThreads(),
                new ThreadFactoryImpl("NettyServerNIOSelector_"));
        }
    }

    private EventLoopGroup buildBossEventLoopGroup() {
        if (useEpoll()) {
            return new EpollEventLoopGroup(1, new ThreadFactoryImpl("NettyEPOLLBoss_"));
        } else {
            return new NioEventLoopGroup(1, new ThreadFactoryImpl("NettyNIOBoss_"));
        }
    }

    private boolean useEpoll() {
        return NetworkUtil.isLinuxPlatform() && nettyServerConfig.isUseEpollNativeSelector() && Epoll.isAvailable();
    }

    private ExecutorService buildPublicExecutor(NettyServerConfig nettyServerConfig) {
        int publicThreadNums = nettyServerConfig.getServerCallbackExecutorThreads();
        if (publicThreadNums <= 0) {
            publicThreadNums = 4;
        }

        return Executors.newFixedThreadPool(publicThreadNums, new ThreadFactoryImpl("NettyServerPublicExecutor_"));
    }

    @ChannelHandler.Sharable
    public class NettyServerHandler extends SimpleChannelInboundHandler<RemotingCommand> {

        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemotingCommand cmd) {
            int command = cmd.getCommand();
            if (nettyServerConfig.isReceiveType()) {
                switch (cmd.getCommandCustomHeader().getReceiveType()) {
                    case DEVICE_REPORT_TYPE:
                        processRequestCommand(channelHandlerContext, cmd);
                        break;
                    case USER_ISSUE_TYPE:
                        processResponseCommand(channelHandlerContext, cmd);
                        break;
                }
            } else {
                // 非服务端发送指令返回数据
                if (!processResponseCommand(channelHandlerContext, cmd)) {
                    processRequestCommand(channelHandlerContext, cmd);
                }
            }
        }
    }

    @ChannelHandler.Sharable
    public class NettyConnectManageHandler extends ChannelDuplexHandler {
        @Override
        public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelRegistered {}", remoteAddress);
            super.channelRegistered(ctx);
        }

        @Override
        public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelUnregistered, the channel[{}]", remoteAddress);
            super.channelUnregistered(ctx);
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelActive, the channel[{}]", remoteAddress);
            super.channelActive(ctx);
            if (NettyRemotingServer.this.channelEventListener != null) {
                NettyRemotingServer.this
                    .putNettyEvent(new NettyEvent(NettyEventType.CONNECT, remoteAddress, ctx.channel()));
            }
        }

        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.info("NETTY SERVER PIPELINE: channelInactive, the channel[{}]", remoteAddress);
            super.channelInactive(ctx);
            if (NettyRemotingServer.this.channelEventListener != null) {
                NettyRemotingServer.this
                    .putNettyEvent(new NettyEvent(NettyEventType.CLOSE, remoteAddress, ctx.channel()));
            }
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent)evt;
                if (event.state().equals(IdleState.ALL_IDLE)) {
                    final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
                    log.warn("NETTY SERVER PIPELINE: IDLE exception [{}]", remoteAddress);
                    RemotingHelper.closeChannel(ctx.channel());
                    if (NettyRemotingServer.this.channelEventListener != null) {
                        NettyRemotingServer.this
                            .putNettyEvent(new NettyEvent(NettyEventType.IDLE, remoteAddress, ctx.channel()));
                    }
                }
            }
            ctx.fireUserEventTriggered(evt);
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            final String remoteAddress = RemotingHelper.parseChannelRemoteAddr(ctx.channel());
            log.warn("NETTY SERVER PIPELINE: exceptionCaught {}", remoteAddress);
            log.warn("NETTY SERVER PIPELINE: exceptionCaught exception.{}", ExceptionUtils.getStackTrace(cause));
            if (NettyRemotingServer.this.channelEventListener != null) {
                NettyRemotingServer.this
                    .putNettyEvent(new NettyEvent(NettyEventType.EXCEPTION, remoteAddress, ctx.channel()));
            }

            RemotingHelper.closeChannel(ctx.channel());
        }
    }

    @Override
    public ExecutorService getCallbackExecutor() {
        return this.publicExecutor;
    }

    public ChannelEventListener getChannelEventListener() {
        return channelEventListener;
    }

    public DeviceManager getDeviceManager() {
        return deviceManager;
    }

    /**
     * 执行一次，无返回
     * @Author wangwei
     * @Date 2024/2/28
     */
    public void invokeOneway(String customerUniqueIdentity,  RemotingCommand request) throws RemotingSendRequestException {
        DeviceLiveInfo deviceInfo = this.deviceManager.getDeviceInfo(customerUniqueIdentity);
        if (deviceInfo == null) {
            throw new BusinessException("设备唯一标识未找到：{}", customerUniqueIdentity);
        }
        Channel channel = deviceInfo.getChannel();
        if (channel != null && channel.isActive()) {
            this.invokeOnewayImpl(channel, request);
        } else {
            this.closeChannel(channel);
            String addr = RemotingHelper.parseChannelRemoteAddr(channel);
            throw new BusinessException("channel未注册或未激活，{}", addr);
        }
    }

    /**
     * 同步执行
     * 
     * @param customerUniqueIdentity 客户端上报的唯一标识
     * @param request 请求内容
     * @return RemotingCommand
     */
    public RemotingCommand invokeSync(String customerUniqueIdentity, RemotingCommand request)
        throws RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
        DeviceLiveInfo deviceInfo = this.deviceManager.getDeviceInfo(customerUniqueIdentity);
        if (deviceInfo == null) {
            throw new BusinessException("设备唯一标识未找到：{}", customerUniqueIdentity);
        }
        Channel channel = deviceInfo.getChannel();
        if (channel != null && channel.isActive()) {
            return this.invokeSyncImpl(channel, request, this.nettyServerConfig.getRequestTimeoutMillis());
        } else {
            this.closeChannel(channel);
            throw new BusinessException("channel未注册或未激活，{}", customerUniqueIdentity);
        }
    }

    /**
     *
     * @param customerUniqueIdentity 客户端上报的唯一标识
     * @param request 请求内容
     * @param requestTimeoutMillis 超时时间
     * @return
     */
    public RemotingCommand invokeSync(String customerUniqueIdentity, RemotingCommand request, Long requestTimeoutMillis)
        throws RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
        DeviceLiveInfo deviceInfo = this.deviceManager.getDeviceInfo(customerUniqueIdentity);
        if (deviceInfo == null) {
            throw new BusinessException("设备唯一标识未找到：{}", customerUniqueIdentity);
        }
        Channel channel = deviceInfo.getChannel();
        if (channel != null && channel.isActive()) {
            return this.invokeSyncImpl(channel, request, requestTimeoutMillis);
        } else {
            this.closeChannel(channel);
            throw new BusinessException("channel未注册或未激活，{}", customerUniqueIdentity);
        }
    }

    /**
     * 异步执行
     * 
     * @param customerUniqueIdentity 客户端上报的唯一标识
     * @param request 请求内容
     * @param invokeCallback
     */
    public void invokeAsync(String customerUniqueIdentity, RemotingCommand request, InvokeCallback invokeCallback)
        throws RemotingConnectException {
        DeviceLiveInfo deviceInfo = this.deviceManager.getDeviceInfo(customerUniqueIdentity);
        if (deviceInfo == null) {
            throw new BusinessException("设备唯一标识未找到：{}", customerUniqueIdentity);
        }
        Channel channel = deviceInfo.getChannel();
        if (channel != null && channel.isActive()) {
            this.invokeAsyncImpl(channel, request, this.nettyServerConfig.getRequestTimeoutMillis(), invokeCallback);
        } else {
            this.closeChannel(channel);
            throw new RemotingConnectException(customerUniqueIdentity);
        }
    }

    /**
     *
     * @param customerUniqueIdentity 客户端上报的唯一标识
     * @param request 请求内容
     * @param invokeCallback 回调内容
     * @param requestTimeoutMillis 超时时间
     * @throws RemotingConnectException
     */
    public void invokeAsync(String customerUniqueIdentity, RemotingCommand request, InvokeCallback invokeCallback,
        Long requestTimeoutMillis) throws RemotingConnectException {
        DeviceLiveInfo deviceInfo = this.deviceManager.getDeviceInfo(customerUniqueIdentity);
        if (deviceInfo == null) {
            throw new BusinessException("设备唯一标识未找到：{}", customerUniqueIdentity);
        }
        Channel channel = deviceInfo.getChannel();
        if (channel != null && channel.isActive()) {
            this.invokeAsyncImpl(channel, request, requestTimeoutMillis, invokeCallback);
        } else {
            this.closeChannel(channel);
            throw new RemotingConnectException(customerUniqueIdentity);
        }
    }

    /**
     * 异步执行
     * 
     * @param customerUniqueIdentity 客户端上报的唯一标识
     * @param request 请求内容
     * @return CompletableFuture
     */
    public CompletableFuture<RemotingCommand> invokeAsyncFuture(String customerUniqueIdentity,
        RemotingCommand request) {
        CompletableFuture<RemotingCommand> future = new CompletableFuture<>();
        try {
            this.invokeAsync(customerUniqueIdentity, request, new InvokeCallback() {
                @Override
                public void operationComplete(ResponseFuture responseFuture) {

                }

                @Override
                public void operationSucceed(RemotingCommand response) {
                    future.complete(response);
                }

                @Override
                public void operationFail(Throwable throwable) {
                    future.completeExceptionally(throwable);
                }
            });
        } catch (RemotingConnectException e) {
            future.completeExceptionally(e);
        }
        return future;
    }

    /**
     * 主动关闭Channel
     */
    public void closeChannel(Channel channel) {
        if (channel != null) {
            // 主动关闭
            String addr = RemotingHelper.parseChannelRemoteAddr(channel);
            log.info("通道未活跃，主动关闭：{}", addr);
            this.deviceManager.remove(channel);
            RemotingHelper.closeChannel(channel);
        } else {
            log.info("通道已关闭");
        }
    }
}
